package org.study.debezium.config;

import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 作用描述
 *
 * @author doveylovey
 * @version v1.0.0
 * @email 1135782208@qq.com
 * @date 2025年01月03日
 */
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
@Configuration
public class DebeziumConfiguration {
    private final MysqlDebeziumConfiguration debeziumConfiguration;

    @Bean
    DebeziumServerBootstrap debeziumServerBootstrap() {
        DebeziumServerBootstrap debeziumServerBootstrap = new DebeziumServerBootstrap();
        DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(debeziumConfiguration.buidProperties())
                .notifying(DebeziumHandler::notifyEvent)
                .build();
        debeziumServerBootstrap.setDebeziumEngine(debeziumEngine);
        return debeziumServerBootstrap;
    }
}
